package com.shujia.mr.kqzl;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.ArrayList;

//<"1001-20230806",60>
//...
public class KqzlMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    private ArrayList<String> idList = new ArrayList<>();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //因为map是按照行记录读取的
        //我们的数据第一行是标题，涵盖了检测点的编号
        String line = value.toString();
        if (line.startsWith("date")) {
            //如果一行以date开头，说明这一行是标题
            String[] infos = line.split(",");
            //编号是从第4列开始的
            for (int i = 3; i < infos.length; i++) {
                idList.add(infos[i].substring(0, infos[i].length() - 1)); // 1001A
            }
        } else {
            String[] infos = line.split(",");
            //判断这一行是否是获取PM2.5的值
            if ("PM2.5".equals(infos[2])) {
                int j = 0;
                for (int i = 3; i < infos.length && j < idList.size(); i++, j++) {
                    if (infos[i] == null || "".equals(infos[i])) {
                        infos[i] = "0";
                    }
                    context.write(new Text(infos[0] + "-" + idList.get(j)), new LongWritable(Long.parseLong(infos[i])));
                }

                if (j != idList.size() - 1) {
                    for (; j < idList.size(); j++) {
                        context.write(new Text(infos[0] + "-" + idList.get(j)), new LongWritable(0L));
                    }
                }

            }
        }
    }
}
